package com.ww.disruptor.consumer;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.ww.disruptor.event.OrderEvent;
import com.ww.disruptor.factory.OrderEventFactory;
import com.ww.disruptor.handler.OrderEventHandler;
import com.ww.disruptor.producer.OrderEventProducer;

import java.nio.ByteBuffer;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class OrderEventDemo03 {

    public static void main(String[] args) {
        // 1、实例化disruptor对象
        OrderEventFactory orderEventFactory = new OrderEventFactory();
        int ringBufferSize = 1024 * 1024;

        ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

        Disruptor<OrderEvent> disruptor = new Disruptor<>(
                orderEventFactory, // 消息event工厂对象
                ringBufferSize, // 容器的长度
                executor, // 线程池（建议使用自定义线程池）
                ProducerType.SINGLE, // 单生产者模式
                new BlockingWaitStrategy() // 等待策略，这里用的阻塞策略
        );

        // 2、添加消费者监听器
        disruptor.handleEventsWith(new OrderEventHandler());

        // 3、启动disruptor
        disruptor.start();

        // 4、获取实际存储数据的容器
        RingBuffer<OrderEvent> ringBuffer = disruptor.getRingBuffer();
        OrderEventProducer producer = new OrderEventProducer(ringBuffer);
        ByteBuffer byteBuffer = ByteBuffer.allocate(8);

        for (int i = 0; i < 100; i++) {
            byteBuffer.putLong(0, i);
            producer.sendData(byteBuffer);
        }

        disruptor.shutdown();
        executor.shutdown();
    }
}
